package com.apes.framework.plugin.mq;


import com.apes.framework.api.SpringManager;
import com.apes.framework.plugin.esb.api.Request;
import com.apes.framework.plugin.esb.processor.ActionProcessor;
import com.apes.framework.plugin.mq.rocketmq.autoconfigure.RocketMQMessageListener;
import com.apes.framework.rop.event.RopListener;
import com.apes.oap.RopContext;
import com.apes.oap.event.AfterStartedRopEvent;
import com.apes.oap.event.RopEventListener;
import com.google.common.eventbus.AsyncEventBus;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.stereotype.Service;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;

/**
 * 功能：MQEvent RopListener
 *
 * @author xul
 * @create 2017-12-05 14:05
 */
@RopListener
public class MQEventListener implements RopEventListener<AfterStartedRopEvent> {
    protected final Logger logger = LogManager.getLogger(MQEventListener.class);

    public void onRopEvent(AfterStartedRopEvent ropRopEvent) {
        RopContext ropContext = ropRopEvent.getRopContext();
        if (SpringManager.getApplicationContext() == null) {
            SpringManager.setApplicationContext(ropContext.getApplicationContext());
        }
        addConsumer(ropContext.getApplicationContext());
        SpringManager.getBean(OutboxEventConsumer.class).initConsumer();
        SpringManager.getBean(AsyncEventBus.class).register(SpringManager.getBean(MessageChannel.class)); /** 单例模式 **/
    }

    private void addConsumer(final ApplicationContext context) throws BeansException {
        String[] beanNames = context.getBeanNamesForAnnotation(Service.class);
        for (final String beanName : beanNames) {
            Class<?> handlerType = context.getType(beanName);
            ReflectionUtils.doWithMethods(handlerType, method -> {
                ReflectionUtils.makeAccessible(method);
                com.apes.framework.plugin.mq.annotation.Consumer annotation = AnnotationUtils.findAnnotation(method, com.apes.framework.plugin.mq.annotation.Consumer.class);
                Consumer consumer = new Consumer() {

                    private RocketMQMessageListener rocketMQMessageListener;

                    @Override
                    public String getTopic() {
                        return annotation.topic();
                    }

                    @Override
                    public boolean isBackup() {
                        return annotation.backup();
                    }

                    @Override
                    public int getPeriod() {
                        return annotation.period();
                    }

                    @Override
                    public boolean isTransaction() {
                        return annotation.transaction();
                    }

                    @Override
                    public RocketMQMessageListener getRocketMQMessageListener() {
                        return this.rocketMQMessageListener;
                    }

                    @Override
                    public void setRocketMQMessageListener(RocketMQMessageListener rocketMQMessageListener) {
                        this.rocketMQMessageListener = rocketMQMessageListener;
                    }

                    @Override
                    public boolean consume(MQEvent ropEvent) {
                        ActionProcessor processor = new ActionProcessor();
                        Object target = context.getBean(beanName);
                        processor.setTarget(target);
                        processor.setMethod(method);
                        ConsumerAdapter adapter = context.getBean(ConsumerAdapter.class);
                        return adapter.consume(processor, ropEvent, isTransaction());
                    }
                };

                SpringManager.getBean(MessageChannel.class).addConsumer(consumer);
            }, method -> !method.isSynthetic()
                    && AnnotationUtils.findAnnotation(method, com.apes.framework.plugin.mq.annotation.Consumer.class) != null
                    && method.getParameterTypes().length == 1
                    && ClassUtils.isAssignable(Request.class, method.getParameterTypes()[0])
                    && AnnotationUtils.findAnnotation(method, com.apes.framework.plugin.mq.annotation.Consumer.class).topic() != null);
        }

        if (context.getParent() != null) {
            addConsumer(context.getParent());
        }
    }

    public int getOrder() {
        return 200;
    }
}

